package cn.iocoder.springboot.lab03.kafkademo.producer;

import cn.iocoder.springboot.lab03.kafkademo.message.Demo02Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

@Component
public class Demo02Producer {

    private Logger LOGGER = LoggerFactory.getLogger(getClass());

    @Resource
    private KafkaTemplate<Object, Object> kafkaTemplate;

    private static int MESSAGE_SIZE = 10000;

    public ListenableFuture<SendResult<Object, Object>> asyncSend() throws ExecutionException, InterruptedException {
        // 创建 Demo02Message 消息
//        Demo02Message message = new Demo02Message();
//        message.setId(id);
        // 异步发送消息
        LOGGER.info("开始发送消息");
        long before = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_SIZE; i++) {
            Demo02Message message = new Demo02Message();
            message.setId(i);
            kafkaTemplate.send(Demo02Message.TOPIC, message).get();
            LOGGER.info("发送条数：{}", i);
        }
        long now = System.currentTimeMillis();
        LOGGER.info("发送花费时间：{}", now - before);
        return null;
    }
}
